package utils

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"mq/config"
	"os"

	"github.com/apache/rocketmq-client-go/v2/primitive"
)

var Myproducer rocketmq.Producer
var Myconsumer rocketmq.PushConsumer
var Reproducer rocketmq.Producer
var Reconsumer rocketmq.PushConsumer

var ShowConsumer rocketmq.PushConsumer
var ShowReConsumer rocketmq.PushConsumer

func init() {
	Myproducer, _ = rocketmq.NewProducer(
		producer.WithNameServer(config.NAMESRV),
		producer.WithRetry(2),
		producer.WithGroupName(config.PRODUCER),
	)
	err := Myproducer.Start()
	if err != nil {
		fmt.Printf("start producer error: %s", err.Error())
		os.Exit(1)
	}

	Myconsumer, err = rocketmq.NewPushConsumer(consumer.WithNameServer(config.NAMESRV),
		consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithGroupName(config.CONSUMER),
	)

	if err != nil {
		fmt.Printf("start comsumer error: %s", err.Error())
		os.Exit(1)
	}

	Reproducer, err = rocketmq.NewProducer(
		producer.WithNameServer(config.NAMESRV),
		producer.WithRetry(2),
		producer.WithGroupName(config.DATA_PRODUCER),
	)
	err = Reproducer.Start()
	if err != nil {
		fmt.Printf("start producer error: %s", err.Error())
		os.Exit(1)
	}

	Reconsumer, err = rocketmq.NewPushConsumer(consumer.WithNameServer(config.NAMESRV),
		consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithGroupName(config.DATA_CONSUMER),
	)

	if err != nil {
		fmt.Printf("start comsumer error: %s", err.Error())
		os.Exit(1)
	}

	ShowReConsumer, err = rocketmq.NewPushConsumer(consumer.WithNameServer(config.NAMESRV),
		consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithGroupName(config.SHOW_CONSUMER),
	)

	if err != nil {
		fmt.Printf("start comsumer error: %s", err.Error())
		os.Exit(1)
	}

	//
	//showConsumer, err = rocketmq.NewPushConsumer(consumer.WithNameServer(config.NAMESRV),
	//	consumer.WithConsumerModel(consumer.Clustering),
	//	consumer.WithGroupName(config.DATA_CONSUMER),
	//)
	//
	//if err != nil {
	//	fmt.Printf("start comsumer error: %s", err.Error())
	//	os.Exit(1)
	//}

}

func SendSyncMessage(tropic string, message []byte, p rocketmq.Producer) {

	// 发送消息
	result, err := p.SendSync(context.Background(), &primitive.Message{
		Topic: tropic,
		Body:  message,
	})

	if err != nil {
		fmt.Printf("Send:\nsend message error: %s\n", err.Error())
	} else {
		fmt.Printf("Send:\nsend message seccess: result=%s\n", result.String())
	}

}

//func SubcribeMessage() {
//	// 订阅主题、消费
//	endPoint := []string{config.NAMESRV}
//	// 创建一个consumer实例
//	c, err := rocketmq.NewPushConsumer(consumer.WithNameServer(endPoint),
//		consumer.WithConsumerModel(consumer.Clustering),
//		consumer.WithGroupName("ConsumerGroupName"),
//	)
//
//	// 订阅topic
//	err = c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
//		for i := range msgs {
//			fmt.Printf("subscribe callback : %v \n", string(msgs[i].Body))
//		}
//		return consumer.ConsumeSuccess, nil
//	})
//
//	if err != nil {
//		fmt.Printf("subscribe message error: %s\n", err.Error())
//	}
//
//	// 启动consumer
//	err = c.Start()
//
//	if err != nil {
//		fmt.Printf("consumer start error: %s\n", err.Error())
//		os.Exit(-1)
//	}
//
//	err = c.Shutdown()
//	if err != nil {
//		fmt.Printf("shutdown Consumer error: %s\n", err.Error())
//	}
//}

//func CreateTopic(topicName string) {
//	endPoint := []string{"192.168.88.128:9876"}
//	// 创建主题
//	testAdmin, err := admin.NewAdmin(admin.WithResolver(primitive.NewPassthroughResolver(endPoint)))
//	if err != nil {
//		fmt.Printf("connection error: %s\n", err.Error())
//	}
//	err = testAdmin.CreateTopic(context.Background(), admin.WithTopicCreate(topicName))
//	if err != nil {
//		fmt.Printf("createTopic error: %s\n", err.Error())
//	}
//}
